package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockIncrBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;

/**
 * 涨跌幅分时窗口函数
 */
public class StockIncrWindowFunction extends RichWindowFunction<CleanBean, StockIncrBean, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<StockIncrBean> out) throws Exception {
        /**
         * 开发步骤：
         * 1.记录最新个股
         * 2.格式化日期
         * 3.指标计算
         *   涨跌、涨跌幅、振幅
         * 4.封装输出数据
         */


        //1.记录最新个股
        CleanBean line = null; //缓存的是上一条数据
        for (CleanBean cleanBean : input) {
            if(line == null){
                line = cleanBean;
            }
            if(line.getEventTime()<cleanBean.getEventTime()){
                line = cleanBean;
            }
        }
        //2.格式化日期
        Long tradeTime = DateUtil.longTimeTransfer(line.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
        String tradeDate = DateUtil.longTimeToString(line.getEventTime(), Constant.format_yyyy_mm_dd);
        /**
         * 涨跌、涨跌幅、振幅
         * 今日涨跌=当前价-前收盘价
         * 今日涨跌幅（%）=（当前价-前收盘价）/ 前收盘价 * 100%
         * 今日振幅 =（当日最高点的价格－当日最低点的价格）/昨天收盘价×100%
         */
        //今日涨跌=当前价-前收盘价
        BigDecimal upDown = line.getTradePrice().subtract(line.getPreClosePrice());
        //今日涨跌幅（%）=（当前价-前收盘价）/ 前收盘价 * 100%
        BigDecimal riseAndFall = upDown.divide(line.getPreClosePrice(), 2, RoundingMode.HALF_UP);
        //今日振幅 =（当日最高点的价格－当日最低点的价格）/昨天收盘价×100%
        BigDecimal amplitude = (line.getHighPrice().subtract(line.getLowPrice())).divide(line.getPreClosePrice(), 2, RoundingMode.HALF_UP);

        // 4.封装输出数据
        out.collect(new StockIncrBean(
                tradeTime,
                line.getSecCode(),
                line.getSecName(),
                riseAndFall,
                line.getTradePrice(),
                upDown,
                line.getTradeVolumn(),
                amplitude,
                line.getPreClosePrice(),
                line.getTradeAmt(),
                tradeDate,
                line.getEventTime(),
                line.getSource()
        ));
    }
}
